package com.share.order.receiver;

import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import com.share.common.core.utils.StringUtils;
import com.share.common.rabbit.constant.MqConst;
import com.share.order.api.domain.EndOrderVo;
import com.share.order.api.domain.SubmitOrderVo;
import com.share.order.service.IOrderInfoService;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.dao.TransientDataAccessException;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * 订单消息接收者
 * 处理与订单相关的消息消费，主要负责处理充电宝租借订单的提交请求
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class OrderReceiver {

    /**
     * 订单信息服务接口，用于处理订单的业务逻辑
     */
    private final IOrderInfoService orderInfoService;

    /**
     * Redis模板，用于实现分布式锁，防止重复处理订单消息
     */
    private final RedisTemplate redisTemplate;

    /**
     * 监听订单提交消息并处理
     * 通过RabbitListener注解绑定到指定的交换机、队列和路由键
     *
     * @param content 消息内容（JSON格式字符串）
     * @param message 原始消息对象，包含消息属性等信息
     * @param channel 消息通道，用于手动确认消息处理状态
     */
    @SneakyThrows // Lombok注解，自动处理异常抛出
    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = MqConst.EXCHANGE_ORDER, durable = "true"), // 绑定的订单交换机，持久化
            value = @Queue(value = MqConst.QUEUE_SUBMIT_ORDER, durable = "true"),   // 绑定的订单提交队列，持久化
            key = MqConst.ROUTING_SUBMIT_ORDER)                                     // 绑定的订单提交路由键
    )
    public void submitOrder(String content, Message message, Channel channel) {
        log.info("[订单服务]接收到租借充电宝消息：{}", content);

        // 将JSON格式的消息内容转换为订单提交对象
        SubmitOrderVo orderForm = JSONObject.parseObject(content, SubmitOrderVo.class);

        // 获取消息编号，用于防重复处理
        String messageNo = orderForm.getMessageNo();

        // 生成Redis分布式锁的键，格式：order:submit:消息编号
        String key = "order:submit:" + messageNo;

        // 使用setIfAbsent实现分布式锁，防止重复处理同一订单请求
        // 若键不存在则设置成功，返回true；若已存在则返回false
        // 锁的过期时间为1小时，避免死锁
        boolean isExist = redisTemplate.opsForValue().setIfAbsent(key, messageNo, 1, TimeUnit.HOURS);

        // 如果锁已存在，说明是重复请求，直接返回
        if (!isExist) {
            log.info("检测到重复的订单提交请求: {}", content);
            return;
        }

        try {
            // 调用订单服务的方法保存订单信息
            orderInfoService.saveOrder(orderForm);

            // 手动确认消息已被成功处理
            // 参数说明：消息标签（唯一标识），是否批量确认
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            log.info("[订单服务]订单提交处理成功，消息编号：{}", messageNo);
        } catch (Exception e) {
            // 处理异常情况，记录错误日志
            log.error("订单服务：订单提交处理失败，消息编号：{}", messageNo, e);

            // 出现异常时删除分布式锁，允许后续重试
            redisTemplate.delete(key);

            // 手动拒绝消息并使其重新入队，以便后续重试
            // 参数说明：消息标签，是否批量处理，是否重新入队
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
    }

    /**
     * 处理结束订单的消息消费者
     * 监听MQ队列，接收充电宝归还消息并处理订单结束逻辑
     */
    @RabbitListener(bindings = @QueueBinding(exchange = @Exchange(value = MqConst.EXCHANGE_ORDER, durable = "true"), value = @Queue(value = MqConst.QUEUE_END_ORDER, durable = "true"), key = MqConst.ROUTING_END_ORDER))
    public void endOrder(String content, Message message, Channel channel) {
        log.info("[订单服务]收到归还充电宝消息: {}", content);
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        // 1. 解析消息内容
        EndOrderVo endOrderVo;
        try {
            endOrderVo = JSONObject.parseObject(content, EndOrderVo.class);
        } catch (Exception e) {
            log.error("解析结束订单消息失败: {}", content, e);
            try {
                // 消息格式错误，拒绝并丢弃消息(不再重试)
                channel.basicNack(deliveryTag, false, false);
            } catch (IOException ioException) {
                log.error("拒绝消息失败: {}", deliveryTag, ioException);
            }
            return;
        }

        // 2. 校验消息完整性
        String messageNo = endOrderVo.getMessageNo();
        if (StringUtils.isEmpty(messageNo)) {
            log.error("结束订单消息编号为空: {}", content);
            try {
                channel.basicNack(deliveryTag, false, false);
            } catch (IOException e) {
                log.error("拒绝消息失败: {}", deliveryTag, e);
            }
            return;
        }

        // 3. 分布式锁实现幂等性
        // 防止同一消息被多次处理(例如重试时)
        String key = "order:endOrder:" + messageNo;
        boolean lockAcquired = false;

        try {
            // 获取分布式锁，设置1小时过期防止死锁
            lockAcquired = redisTemplate.opsForValue().setIfAbsent(key, messageNo, 1, TimeUnit.HOURS);

            if (!lockAcquired) {
                log.info("重复处理的结束订单消息: {}", messageNo);
                // 已处理过的消息直接确认，不再处理
                channel.basicAck(deliveryTag, false);
                return;
            }

            // 4. 处理订单结束业务逻辑
            log.info("开始处理结束订单，消息编号: {}", messageNo);
            orderInfoService.endOrder(endOrderVo);

            // 5. 手动确认消息
            // 确认消息处理成功，从队列中移除
            channel.basicAck(deliveryTag, false);
            log.info("结束订单处理成功，消息编号: {}", messageNo);

        } catch (IOException e) {
            // 业务异常(如余额不足、订单状态异常)
            log.error("结束订单业务处理失败，消息编号: {}", messageNo, e);
            try {
                // 拒绝消息并重试(可能需要人工干预)
                channel.basicNack(deliveryTag, false, true);
            } catch (IOException ioException) {
                log.error("消息重试失败，消息编号: {}", messageNo, ioException);
            }

        } catch (Exception e) {
            // 系统异常(如数据库连接失败)
            log.error("结束订单系统处理失败，消息编号: {}", messageNo, e);
            try {
                // 拒绝消息并重试
                channel.basicNack(deliveryTag, false, true);
            } catch (IOException ioException) {
                log.error("消息重试失败，消息编号: {}", messageNo, ioException);
            }

        } finally {
            // 6. 确保锁的正确释放
            // 特殊情况: 通道关闭但锁未释放时，主动释放锁
            if (lockAcquired && !channel.isOpen()) {
                try {
                    redisTemplate.delete(key);
                    log.info("通道关闭，释放分布式锁: {}", key);
                } catch (Exception e) {
                    log.error("释放锁失败: {}", key, e);
                }
            }
        }
    }

    /**
     * 监听订单支付成功消息：更新订单状态；扣减商品库存
     *
     * @param orderNo 订单编号
     * @param message RabbitMQ消息对象
     * @param channel RabbitMQ通道，用于手动确认消息
     */
    @SneakyThrows
    @RabbitListener(bindings = @QueueBinding(exchange = @Exchange(value = MqConst.EXCHANGE_PAYMENT_PAY, durable = "true"), value = @Queue(value = MqConst.QUEUE_PAYMENT_PAY, durable = "true"), key = MqConst.ROUTING_PAYMENT_PAY))
    public void processPaySucess(String orderNo, Message message, Channel channel) {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        log.info("[订单服务]收到支付成功消息，订单号：{}，投递标签：{}", orderNo, deliveryTag);

        try {
            // 校验订单号有效性
            if (StringUtils.isEmpty(orderNo)) {
                log.error("[订单服务]收到无效订单号，丢弃消息。投递标签：{}", deliveryTag);
                channel.basicAck(deliveryTag, false);
                return;
            }

            // 记录消息处理开始
            log.info("[订单服务]开始处理支付成功消息，订单号：{}", orderNo);

            // 处理支付成功业务逻辑
            orderInfoService.processPaySucess(orderNo);

            // 记录业务处理完成
            log.info("[订单服务]订单支付状态已更新，订单号：{}", orderNo);

            // 手动确认消息
            channel.basicAck(deliveryTag, false);
            log.info("[订单服务]消息确认成功，订单号：{}，投递标签：{}", orderNo, deliveryTag);

        } catch (Exception e) {
            // 记录异常信息，包含订单号和投递标签
            log.error("[订单服务]处理支付成功消息失败，订单号：{}，投递标签：{}", orderNo, deliveryTag, e);

            // 判断是否需要重试
            boolean requeue = shouldRequeue(e);

            // 拒绝消息并根据配置决定是否重新入队
            channel.basicNack(deliveryTag, false, requeue);

            if (requeue) {
                log.info("[订单服务]消息已重新入队，订单号：{}，投递标签：{}", orderNo, deliveryTag);
            } else {
                log.warn("[订单服务]消息已丢弃，订单号：{}，投递标签：{}", orderNo, deliveryTag);
            }
        }
    }

    /**
     * 根据异常类型判断是否需要重新入队
     *
     * @param e 异常对象
     * @return 需要重试返回true，否则返回false
     */
    private boolean shouldRequeue(Exception e) {
        // 可根据具体异常类型决定是否重试
        return e instanceof TransientDataAccessException;
    }
}